package kafka

import (
	"github.com/sirupsen/logrus"
	event2 "go_knowledge/mq/kafka/event"
	consumer2 "go_knowledge/mq/kafka/event/kafka/consumer"
)

// 根据事件自动监听对应TOPIC
type autoTopicListener struct {
	addr   []string
	logger *logrus.Logger
}

func NewAutoTopicListener(addr []string) *autoTopicListener {
	return &autoTopicListener{addr: addr}
}

func (atl *autoTopicListener) SetLogger(logger *logrus.Logger) {
	atl.logger = logger
}

func (atl *autoTopicListener) Listen(group string, event event2.PbEvent, eventHandler event2.Handler) {

	queueName, err := event2.MakeTopic(event)
	if nil != err {
		atl.logger.Errorf("make topic by event failed: %s", err)
		return
	}

	client := consumer2.NewDefaultConsumerClient(atl.addr, queueName, group)
	if nil != atl.logger {
		client.SetLogger(atl.logger)
	}

	client.Handle(eventHandler)
}
